package flink;

import com.alibaba.fastjson.JSON;
import flink.model.UserAction;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Properties;

/**
 * Created by liujian on 2021/1/28.
 */
public class WindowUserAction {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);


        Properties p = new Properties();
        p.setProperty("bootstrap.servers", "192.168.62.36:6667");
        p.setProperty("group.id", "test");

        final DataStreamSource<String> dataStream = env.addSource(
                new FlinkKafkaConsumer<String>("flink_out", new SimpleStringSchema(), p)
        );

        dataStream.print();

        dataStream.map(new MapFunction<String, UserAction>() {
            public UserAction map(String s) throws Exception {
                return JSON.parseObject(s, UserAction.class);
            }
        }).assignTimestampsAndWatermarks(
                WatermarkStrategy.<UserAction>forBoundedOutOfOrderness(
                        Duration.ofSeconds(3))
                        .withTimestampAssigner(new SerializableTimestampAssigner<UserAction>() {
                            public long extractTimestamp(UserAction userAction, long l) {
                                SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");
                                try {
                                    return dateFormat.parse(userAction.getUserActionTime()).getTime();
                                } catch (ParseException e) {
                                    e.printStackTrace();
                                }
                                return 0;
                            }
                        })

        ).map(new MapFunction<UserAction, Tuple3<Integer, String, Integer>>() {

            @Override
            public Tuple3<Integer, String, Integer> map(UserAction userAction) throws Exception {
                return new Tuple3<>(userAction.getUserId(), userAction.getUserActionTime(), 1);
            }
        }).keyBy(new KeySelector<Tuple3<Integer, String, Integer>, Integer>() {
            @Override
            public Integer getKey(Tuple3<Integer, String, Integer> value) throws Exception {
                return value.f0;
            }
        }).window(EventTimeSessionWindows.withGap(Time.minutes(1)))// 设置session windows会话窗口，Gap time会话时间设置为1min，关于session window的概念见官方文档
                .allowedLateness(Time.seconds(10))// 允许延迟的时间，比如，上一条数据是 10:00:00，后面来了一条09:59:50的数据，仍然会在window中计算
                .reduce((ReduceFunction<Tuple3<Integer, String, Integer>>) (value1, value2) -> new Tuple3<>(value1.f0, value1.f1, value1.f2 + value2.f2))
                .filter((FilterFunction<Tuple3<Integer, String, Integer>>) value -> value.f2 < 2)
                .print();

        // 调用execute开始执行
        env.execute("WindowUserAction");


    }


}
